package com.momo.producer;

import org.apache.kafka.clients.producer.*;

import java.util.ArrayList;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @author ：朱博
 * @description：TODO
 * @date ：2021/3/21 14:50
 */
public class KafkaTestWithInterceptor {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();

        //kafka 集 群 ，broker-list
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.0.105:9092");

        /*
         * 序列化配置
         * */
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");


        /**
         * 添加拦截器
         */
        ArrayList<String> interceptor = new ArrayList<>();
        interceptor.add("com.momo.interceptor.CounterInterceptor");
        interceptor.add("com.momo.interceptor.TimeInterceptor");
        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,interceptor);

        /**
         * 自定义分区类的全类限定名
         */
      //props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.momo.partitioner.MyPartitioner");

        /**
         * 生产对象t
         */
        Producer<String, String> producer = new KafkaProducer<String, String>(props);

        for (int i = 0; i < 100; i++) {
            Future<RecordMetadata> mytopic = producer.send(
                    new ProducerRecord<String, String>(
                    "mycallback",
                    //"myIntention",//kafak会根据这个对象的Hash值取模运算选择分区，这里key写死的话就只放一个分区了
                    "LoVeToU"),
                    //回调函数 发送完成的回调相关消息
                    (x,y) -> {
                        if (y == null) {
                            System.out.println("分区-->" + x.partition());
                            System.out.println("发送的偏移量大小-->" + x.offset());
                        }
                    });
            RecordMetadata recordMetadata = mytopic.get();//调用 Future的Get方法 会导致  线程 阻塞 就会实现了 同步方法
            System.err.println(mytopic.isDone());
            String topic = recordMetadata.topic();
            System.out.println("阻塞返回值后的分区"+topic);
        }
        producer.close();
    }
}
